package datasource

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010

object KafkaStream {
  /**
    * 从kafka中获取数据源
    * @param bootstrap
    * @param zookeeper
    * @param topic
    * @param env
    * @return
    */
  def getKafkaStream(bootstrap: String,
                     zookeeper: String,
                     topic: String,
                     env: StreamExecutionEnvironment): DataStream[String] = {
    // 创建连接参数
    val properties = new Properties
    properties.setProperty("bootstrap.servers", bootstrap)
    properties.setProperty("zookeeper.connect", zookeeper)
    properties.setProperty("group.id", "test")
    properties.setProperty("input-data-topic", topic)
    env.addSource(new FlinkKafkaConsumer010[String](topic, new SimpleStringSchema(), properties))
  }

}
